#include <iostream>
#include <vector>
#include <map>
#include <fstream>
#include <cstdlib>
#include <sstream>
#include <queue>
#include <cstring>
#include "rdf3xload.cpp"
#include <unordered_set>
#include <mpi.h>
#include <sys/time.h>
#include <algorithm>
#include <vector>
#include <exception>
#include <climits>

//---------------------------------------------------------------------------
//// RDF-3X
//// (c) 2008 Thomas Neumann. Web site: http://www.mpi-inf.mpg.de/~neumann/rdf3x
////
//// This work is licensed under the Creative Commons
//// Attribution-Noncommercial-Share Alike 3.0 Unported License. To view a copy
//// of this license, visit http://creativecommons.org/licenses/by-nc-sa/3.0/
//// or send a letter to Creative Commons, 171 Second Street, Suite 300,
//// San Francisco, California, 94105, USA.
////---------------------------------------------------------------------------
//
using namespace std;
bool runMPI = false;

bool smallAddressSpace()
	// Is the address space too small?
{
	return sizeof(void*)<8;
}


class MyVertex {
	public:
		MyVertex(string s) {
			name = s;
			outgoingBelongsToPartition= -1;
			incomingBelongsToPartition= -1;
		}

		string getName() {
			return name;
		}

		int outgoingBelongsToPartition;
		int incomingBelongsToPartition;
		string name;

};

class Edge {
	public:
		long int destVertexId;
		long int edgeId;

		Edge();
		Edge(long int destId, long int id) {
			destVertexId = destId;
			edgeId = id;
		}

};

class DataPartitioner {

	bool includeOutEdges = true;	// true => n-out-hop guarantee
	bool includeInEdges = true;	// true => n-in-hop guarantee
	char *inputFileName;
	ifstream in;
	int hopLimit = 2;	// Sets number of hops in n-hop guarantee
	int maxDegree = 15000;	// Sets threshold for ignoring high degree vertices
	long int inputTripleCount = 0;	
	long int totalSubObjCount = 1;	// Gives the number of vertices in the input graph
	int totalPredCount = 1;		// Gives the unique number of edges
	long int partitionSize = 0;		// Gives the approximate size of a partition
	map <string, long int> subObjMap, predMap;	
	map <long int, MyVertex> revSubObjMap;
        map <long int, string>	revPredMap;	
	map <long int, vector<Edge> > graph; 	// Stores the input graph in adjacency list format
	queue <long int> highDegreeVertexList; // Used for creating a separate partition with high degree nodes

	void addToQueue(queue <long int> *q, long int id);
	void addTripleToFile(long int vertexId, long int edgeId, 
			long int destVertexId, ofstream &file, 
			unordered_set <string> &tripleInPartitionMap,  
			bool isHighDegreePartition);
	void iterateOutgoingEdges(long int currentVertex,	
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
		unordered_set <string> &tripleInPartitionMap, int currentHop, 
		queue <long int> *nextLevel,  bool isHighDegreePartition);
	void iterateIncomingEdges(long int currentVertex,	
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
		unordered_set <string> &tripleInPartitionMap, int currentHop, 
		queue <long int> *nextLevel,  bool isHighDegreePartition);
	void n_hop_guarantee(long int vertexId, 
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
		unordered_set <string> &tripleInPartitionMap, bool isHighDegreePartition);
	bool isHighDegreeVertex(long int vertexId);

	public:
		DataPartitioner(char *file);
		bool createInputStream();
		bool parseAndCreateGraph();
		void createPartitions();
		void printGraphToFile();
		void getGraphDetails();
		void getHighDegreeNodes();
};

//Constructor
DataPartitioner::DataPartitioner(char *file) {
	inputFileName = file;
}

bool DataPartitioner::createInputStream() {

	in.open(inputFileName);
	if(!in.is_open()) {
		cerr << "Unable to open " << inputFileName << endl;
		return false;
	}
	return true;
}

// Creates a graph and several maps from the given set of triples
bool DataPartitioner::parseAndCreateGraph() {

	if (!createInputStream())
	       return false;	

	string subject, predicate, object, dot=".";
	typedef vector <Edge> QueueDef;
	typedef map <long int, QueueDef> graphStd;
	while(!in.eof()) {
		getline(in, subject, ' ');

		// Used for stats
		inputTripleCount++;
		//print line number which are power of 100
		/*
		long int lineNumb = 100; 
		while(lineNumb <= inputTripleCount) {
			if(inputTripleCount == lineNumb)
				cout << "Scanning line " << inputTripleCount << endl;
			lineNumb *=100;
		} */
		
		if(subject == "")
			break;
		getline(in, predicate, ' ');
		if(predicate == "")
			return false;
		getline(in, object);
		if(object == "")
			return false;

		char *tmpObj = new char[object.length() + 1];
		strcpy(tmpObj, object.c_str());
		object = object.substr(0, object.length()-2);
/*		if(object.length() > 0) {
			if(tmpObj[0] == '"') {
				tmpObj[0] = '<';
				
				char *ch = "\"";
				size_t found = object.find('>');
				if(found != string::npos)
					ch = ">";
				
				char *tok = strtok(tmpObj, ch);
				if(tok[strlen(tok) - 1] == '\\' && tok[strlen(tok) - 2] != '\\') {
					tok[strlen(tok) -1] = '>';
					object.assign(tok);
				} else {
					object.assign(tok);
					object += ">";
				}
			} else if(tmpObj[0] == '<'){
				char *tok = strtok(tmpObj, ">");
				if(tok[strlen(tok) - 1] == '\\' && tok[strlen(tok) - 2] != '\\') {
					tok[strlen(tok) -1] = '>';
					object.assign(tok);
				} else { 
					object.assign(tok);
					object += ">";
				}
			}
		}
*/
	//	cout << subject << " " << predicate << " " << object << " " << dot << endl;
		
		if(subObjMap.find(subject) == subObjMap.end()) {
			revSubObjMap.insert(map <long int, MyVertex>::value_type(totalSubObjCount, MyVertex(subject)));
			subObjMap[subject] = totalSubObjCount++;
		}
		
		if(subObjMap.find(object) == subObjMap.end()) {
			revSubObjMap.insert(map <long int, MyVertex>::value_type(totalSubObjCount, MyVertex(object)));
			subObjMap[object] = totalSubObjCount++;
		}

		if(predMap.find(predicate) == predMap.end()) {
			revPredMap[totalPredCount] = predicate;	
			predMap[predicate] = totalPredCount++;
		}
		
		long int subId = subObjMap[subject], objId = subObjMap[object], predId = predMap[predicate];
		//If graph doesnt have this vertex add the vertex with empty vector
		if(graph.find(subId) == graph.end()) {
			vector <Edge> q;
			q.push_back(Edge(objId, predId));
			graph.insert(graphStd::value_type(subId, q));

		} else {
			graph[subId].push_back(Edge(objId, predId));
		}
		
		if(graph.find(-1*objId) == graph.end()) {
			vector <Edge> q;
			q.push_back(Edge(subId, predId));
			graph.insert(graphStd::value_type(-1*objId, q));

		} else {
			graph[-1*objId].push_back(Edge(subId, predId));
		}

	}
	in.close();

	totalSubObjCount--;
	totalPredCount--;
	return true;
}

bool DataPartitioner::isHighDegreeVertex(long int vertexId) {
	if(graph[vertexId].size() > this->maxDegree || 
			graph[-1*vertexId].size() > this->maxDegree)
		return true;
	return false;
}

// Adds triple to a partition if its not already added
void DataPartitioner::addTripleToFile(long int vertexId, long int edgeId, long int destVertexId, ofstream &file, 
		unordered_set <string> &tripleInPartitionMap, bool isHighDegreePartition) {
	
	string sub = revSubObjMap.find(vertexId)->second.name;
	string obj = revSubObjMap.find(destVertexId)->second.name;
	string pred = revPredMap[edgeId];

	if(!isHighDegreePartition && (isHighDegreeVertex(vertexId) || isHighDegreeVertex(destVertexId)) )
		return;

	// Using (subject,predicate,object) as the key
	ostringstream tripleKey;
	tripleKey << vertexId << "," << edgeId << "," << destVertexId;

	// Add triple to a partition if it is not already added
	if( tripleInPartitionMap.find(tripleKey.str()) == tripleInPartitionMap.end()) {
		file << sub << " " << pred << " " << obj << " ." <<"\n";
		tripleInPartitionMap.insert(tripleKey.str()); 

		// Used for calculating average replication ratio among partitions
		partitionSize++;
	}
}

void DataPartitioner::addToQueue(queue <long int> *q, long int id) {
	q->push(id);
}

void DataPartitioner::iterateOutgoingEdges(long int currentVertex,
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
		unordered_set <string> &tripleInPartitionMap, int currentHop, 
		queue <long int> *nextLevel, bool isHighDegreePartition) {
	if(!includeOutEdges)
		return;
	long int edgeCount = 0;
	if(graph.find(currentVertex) != graph.end())
		edgeCount = graph[currentVertex].size();
	//	cout << "In partition " << partId << " outgoing edge count is " << edgeCount << endl;	

	for(long int j=0; j<edgeCount; j++) {
		long int destVertexId = graph[currentVertex][j].destVertexId;
		long int edgeId = graph[currentVertex][j].edgeId;

		addTripleToFile(currentVertex, edgeId, destVertexId, partitionOutputStream, 
				tripleInPartitionMap, isHighDegreePartition);

		// (hopLimit - vertexLevel[destVertexId]) indicates the number of hops previsouly
		// visited from that vertexId.
		// Therefore, we add that vertex iff previously it has covered lesser number
		// of hops. 
		if((vertexLevel.count(destVertexId) == 0 || vertexLevel.count(destVertexId) != 0 && 
					currentHop < vertexLevel[destVertexId]) && 
				currentHop != hopLimit) { 
			addToQueue(nextLevel, destVertexId);
			vertexLevel[destVertexId] = currentHop;
		}
	}
}

void DataPartitioner::iterateIncomingEdges(long int currentVertex, 
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
		unordered_set <string> &tripleInPartitionMap, int currentHop, 
		queue <long int> *nextLevel, bool isHighDegreePartition) {
	if(!includeInEdges)
		return;
	long int edgeCount = 0;
	if(graph.find(-1*currentVertex) != graph.end())
		edgeCount = graph[-1*currentVertex].size();
	//	cout << "In partition " << partId << " incoming edge count is " << edgeCount << endl;	

	// iterating throught all the incoming edges of currentVertex
	for(long int j=0; j<edgeCount; j++) {
		long int destVertexId = graph[-1*currentVertex][j].destVertexId;
		long int edgeId = graph[-1*currentVertex][j].edgeId;

		addTripleToFile(destVertexId, edgeId, currentVertex, partitionOutputStream, 
				tripleInPartitionMap, isHighDegreePartition);
		// (hopLimit - vertexLevel[destVertexId]) indicates the number of hops previsouly
		// visited from that vertexId.
		// Therefore, we add that vertex iff previously it has covered lesser number
		// of hops. 
		if((vertexLevel.count(destVertexId) == 0 || vertexLevel.count(destVertexId) != 0 && 
					currentHop < vertexLevel[destVertexId]) && 
				currentHop != hopLimit) { 
			addToQueue(nextLevel, destVertexId);
			vertexLevel[destVertexId] = currentHop;
		}
	}
}

void DataPartitioner::n_hop_guarantee(long int vertexId, 
		map <long int, char> &vertexLevel, ofstream &partitionOutputStream, 
			unordered_set <string> &tripleInPartitionMap, bool isHighDegreePartition) {
	if(!isHighDegreePartition && isHighDegreeVertex(vertexId))
		return;

	queue <long int> *q = new queue <long int> (); 
	addToQueue(q, vertexId);

	int currentHop = 0;
	vertexLevel[vertexId] = currentHop;

	// n-hop guarantee
	while(currentHop != this->hopLimit) {	
		currentHop++;

		queue <long int> *nextLevel = new queue <long int> ();
		// Getting all the vertices that belong to the next hop
		while(q->size()) {

			long int currentVertex = q->front();
			q->pop();

			iterateOutgoingEdges(currentVertex, vertexLevel, partitionOutputStream, 
					tripleInPartitionMap, currentHop, nextLevel, isHighDegreePartition);
			iterateIncomingEdges(currentVertex, vertexLevel, partitionOutputStream, 
					tripleInPartitionMap, currentHop, nextLevel, isHighDegreePartition);
		}
		// NextLevel becomes the new currentLevel in BFS
		q = nextLevel;
	}
}
void DataPartitioner::createPartitions() {

	long int vertexCount = totalSubObjCount;

	//Partition Id
	int numPartitions = 20; 
	int partId = numPartitions - 1;	
	long int vertexId = partId;
	if(runMPI) {
		partId = MPI::COMM_WORLD.Get_rank( ) + 1;	
		vertexId = partId;
		numPartitions = MPI::COMM_WORLD.Get_size ( ); 
	}

//	cout << "Number of partitions = " << numPartitions << endl;

	ostringstream partitionName;

	// Name of the partition
	partitionName << inputFileName << partId;

	// Output stream for the partition
	ofstream partitionOutputStream;
	partitionOutputStream.open(partitionName.str().c_str());

	// Indiacates the partition to which the triple belongs
	// Triple indicated by subId + "," + predId + "," + objId
	unordered_set <string> tripleInPartitionMap;

	// Used to identify the number of hops from the current 
	// vertex that are already included in the partition
	// <vertexId, currentHop>
	map <long int, char> vertexLevel; 

	if(partId != numPartitions-1) {
		//	cout << "For part = " << partId << " vertex range is from " << vertexId << " to " << vertRange << endl;

		// Iterating through all the vertices in a partition
		for(long int i=1; vertexId <= vertexCount; i++) {
			//	cout << "partId = " << partId << " vertexId = " << vertexId << "\n";
			//	Ignoring the high degree vertices
			if(!isHighDegreeVertex(vertexId))
				n_hop_guarantee(vertexId, vertexLevel, 
						partitionOutputStream, tripleInPartitionMap, false);
			// Assuming 1 less partition because last task is used for high-degree-nodes
			vertexId = (numPartitions-1)*i + partId;
		}
	} else {
		// last MPI task is used to create a partition for high degree nodes
		 cout << "High Degree vertex list size = " << highDegreeVertexList.size() << endl;
		 cout << "MaxDegree is " << this->maxDegree << endl;
		while(highDegreeVertexList.size()) {
			long int vertexId = highDegreeVertexList.front();
			highDegreeVertexList.pop();
			n_hop_guarantee(vertexId, vertexLevel, 
					partitionOutputStream, tripleInPartitionMap, true);
		}
	}	
	partitionOutputStream.close();

	ostringstream dbname;
	dbname << inputFileName << partId << "db";
	cout << "dbName added to list is " << dbname.str().c_str() << endl;

	// Loading partition into database		
	struct timeval tim;
	double t1,t2;
	gettimeofday (&tim, NULL);
	t1=tim.tv_sec+(tim.tv_usec/1000000.0);

	rdf3xload_serial(dbname.str().c_str(), partitionName.str().c_str());

	gettimeofday (&tim, NULL);
	t2=tim.tv_sec+(tim.tv_usec/1000000.0);
	printf("%.10lf ms -- time taken to load partitions \n\n", (t2-t1)*1000);

//	printf("Approximate Replication Ratio is %.3lf \n", partitionSize*numPartitions*1.0/inputTripleCount);
}

void DataPartitioner::printGraphToFile() {
	ofstream out;
	out.open("graphTriples");

	for(map<long int, vector <Edge> >::iterator iter = graph.begin(); iter != graph.end(); ++iter) {
		long int vertexId = iter->first;
		if(vertexId < 0)
			continue;
		long int edgeCount = graph[vertexId].size();
		for(long int i=0; i<edgeCount; i++) {
			long int destVertexId = graph[vertexId][i].destVertexId;
			long int edgeId = graph[vertexId][i].edgeId;

			out << vertexId << " " << edgeId << " " << destVertexId << endl;
		}
	}
	out.close();	
}

void DataPartitioner::getGraphDetails() {
	long int totalInDegreeCount = 0;
	long int totalOutDegreeCount = 0;
	map<long int, long int> inDegreeDist;
	map<long int, long int> outDegreeDist;
	for(map<long int, vector <Edge> >::iterator iter = graph.begin(); iter != graph.end(); ++iter) {
		long int vertexId = iter->first;
		long int degree = iter->second.size();
		if(vertexId > 0) {
			totalOutDegreeCount += degree;
			if(outDegreeDist.count(degree))
				outDegreeDist[degree] += 1;
			else
				outDegreeDist[degree] = 1;
		} else {
			totalInDegreeCount += degree;
			if(inDegreeDist.count(degree))
				inDegreeDist[degree] += 1;
			else
				inDegreeDist[degree] = 1;
		}
	}

	cout << "Input triples count = " << inputTripleCount << endl;
	cout << "Average Out-Degree of Graph is " << totalOutDegreeCount*1.0/totalSubObjCount << endl;
	cout << "Average In-Degree of Graph is " << totalInDegreeCount*1.0/totalSubObjCount << endl;
	cout << "Average Degree of the graph is " << (totalOutDegreeCount + totalInDegreeCount)*1.0/totalSubObjCount << endl;
	cout << "Total subject/object count is " << totalSubObjCount << endl;
	cout << "Total predicate count is " << predMap.size() << endl;

	int numb_parts = 20;
	if(runMPI) MPI_Comm_size(MPI_COMM_WORLD, &numb_parts);
	printf("Approximate Replication Ratio is %.3lf \n", partitionSize*numb_parts*1.0/inputTripleCount);

	ofstream inDegDistFile;
	inDegDistFile.open("inDegree_distribution");
	for(map<long int, long int >::iterator iter = inDegreeDist.begin(); iter != inDegreeDist.end(); ++iter) {
		inDegDistFile << iter->first << " " << iter->second << endl;
	}
	inDegDistFile.close();

	ofstream outDegDistFile;
	outDegDistFile.open("outDegree_distribution");
	for(map<long int, long int >::iterator iter = outDegreeDist.begin(); iter != outDegreeDist.end(); ++iter) {
		outDegDistFile << iter->first << " " << iter->second << endl;
	}
	outDegDistFile.close();
}

static bool compareNodeDegree(pair<long int, vector <Edge> > node1, pair< long int, vector <Edge> > node2) {
	long int count1 = node1.second.size();
	long int count2 = node2.second.size();

	return count1 > count2;
}

void DataPartitioner::getHighDegreeNodes() {
	vector <pair<long int, vector <Edge> > > myVec(graph.begin(), graph.end());
	sort(myVec.begin(), myVec.end(), &compareNodeDegree);

	long int count = 0;
	for( vector< pair<long int, vector <Edge> > >::iterator i = myVec.begin(); i != myVec.end(); ++i) {
		count++;

		if(i->second.size() < this->maxDegree)
	//	if(count == 150)
			return;

		long int vertexId = i->first;
		highDegreeVertexList.push(abs(vertexId));
/*		cout << count << " -- ";
		if(vertexId > 0) { 
			cout << "The Out-Degree of -- " << revSubObjMap.find(vertexId)->second.name << "  -- is " << i->second.size() << endl;
		} else {
			cout << "The In-Degree of -- " << revSubObjMap.find(-1*vertexId)->second.name << "  -- is " << i->second.size() << endl;
		}
*/
	}
	
}

int main(int argc,char* argv[])
{
	if (smallAddressSpace())
		cerr << "Warning: Running RDF-3X on a 32 bit system is not supported and will fail for large data sets. Please use a 64 bit system instead!" << endl;

	if (argc<2) {
		cerr <<  "usage: " << argv[0] << " [inputFile] " << endl
			<< "without input file data is read from stdin" << endl
			<< "number of partitions is the number of cores requested" << endl;
		return 1;
	}

	cerr << "RDF-3X turtle importer" << endl
		<< "(c) 2008 Thomas Neumann. Web site: http://www.mpi-inf.mpg.de/~neumann/rdf3x" << endl;

	int partId = 0, numPartitions = 10;
	if(runMPI) {
		MPI::Init (argc, argv);
		partId = MPI::COMM_WORLD.Get_rank( );	
		numPartitions = MPI::COMM_WORLD.Get_size ( ); 
	}

	DataPartitioner dataPartitioner(argv[1]);
	
	struct timeval tim;
	double t1,t2;
	if(partId == 0) {
		gettimeofday (&tim, NULL);
		t1=tim.tv_sec+(tim.tv_usec/1000000.0);
	}
	
	bool noError = dataPartitioner.parseAndCreateGraph();
	if(!noError) {
		cerr << "Error in parsing input file" << endl;
		if(runMPI) MPI_Finalize();
		return 1;
	}
	
	if(partId == 0) {
		gettimeofday(&tim, NULL);
		t2=tim.tv_sec+(tim.tv_usec/1000000.0);
		printf("%.10lf ms - for Graph Creation\n\n", (t2-t1)*1000);
		
		// For stats
		dataPartitioner.printGraphToFile();
		dataPartitioner.getGraphDetails();
		
		gettimeofday (&tim, NULL);
		t1=tim.tv_sec+(tim.tv_usec/1000000.0);
	}
	if(partId == numPartitions - 1) 
		dataPartitioner.getHighDegreeNodes();
	
//	dataPartitioner.createPartitions();
	
	// Making sure all tasks are finished before calcuating the time
	if(runMPI) MPI_Barrier(MPI_COMM_WORLD);
	if(partId == 0) {
		gettimeofday (&tim, NULL);
		t2=tim.tv_sec+(tim.tv_usec/1000000.0);
		printf("%.10lf ms -- time taken to create and load partitions \n\n", (t2-t1)*1000);
	}

	if(runMPI) MPI_Finalize();
	return 0;
}

